package com.azh.server.datam.pagecache;

import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import com.azh.server.common.AbstractQuoteCache;
import com.azh.server.datam.page.Page;
import com.azh.server.datam.page.PageImpl;
import com.azh.server.util.ExceptionUtils;

/**
 * @author zhaozhenhang <zhaozhenhang@kuaishou.com>
 * Created on 2022-04-25
 */
public class PageCacheImpl extends AbstractQuoteCache<Page> implements PageCache{

    private static final int mem_min_lim = 10;

    public static final String db_suffix = ".db";

    private RandomAccessFile file;
    private FileChannel channel;
    private Lock lock;

    // 记录当前打开的数据库文件有多少页
    private AtomicInteger pageNumber;


    public PageCacheImpl(RandomAccessFile file, FileChannel channel, int maxResource) {
        super(maxResource);
        if (maxResource < mem_min_lim) {
            ExceptionUtils.getMessage(new RuntimeException("Memory too small!"));
        }
        long len = 0;
        try {
            len = file.length();
        } catch (IOException e) {
            ExceptionUtils.getMessage(e);
        }
        this.file = file;
        this.channel = channel;
        this.lock = new ReentrantLock();
        this.pageNumber = new AtomicInteger((int) len / PAGE_SIZE);
    }

    @Override
    public int newPage(byte[] initData) {
        int pageNumber = this.pageNumber.incrementAndGet();
        Page page = new PageImpl(pageNumber, initData, null);
        flush(page);
        return pageNumber;
    }

    @Override
    public Page getPage(int pageNumber) throws Exception {
        return get(pageNumber);
    }

    @Override
    public void close() {
        super.close();
        try {
            channel.close();
            file.close();
        } catch (IOException e) {
            ExceptionUtils.getMessage(e);
        }
    }

    @Override
    protected Page getForDB(long key) throws Exception {
        int pageNumber = (int) key;
        long offset = PageCacheImpl.pageOffset(pageNumber);
        ByteBuffer buffer = ByteBuffer.allocate(PAGE_SIZE);
        lock.lock();
        try {
            channel.position(offset);
            channel.read(buffer);
        } catch (IOException e) {
            ExceptionUtils.getMessage(e);
        }
        lock.unlock();
        return new PageImpl(pageNumber, buffer.array(), this);
    }

    @Override
    protected void releaseForDB(Page page) {
        if (page.isDirty()) {
            flush(page);
            page.setDirty(false);
        }
    }

    @Override
    public void release(Page page) {
        release(page.getPageNumber());
    }

    @Override
    public void truncateBytePageNumber(int maxPageNumber) {
        long size = pageOffset(maxPageNumber + 1);
        try {
            file.setLength(size);
        } catch (IOException e) {
            ExceptionUtils.getMessage(e);
        }
        pageNumber.set(maxPageNumber);
    }

    @Override
    public int getPageNumber() {
        return pageNumber.intValue();
    }

    @Override
    public void flushPage(Page page) {
        flush(page);
    }

    private void flush(Page page) {
        int pageNumber = page.getPageNumber();
        long offset = pageOffset(pageNumber);
        lock.lock();
        try {
            ByteBuffer buffer = ByteBuffer.wrap(page.getData());
            channel.position(offset);
            channel.write(buffer);
            channel.force(false);
        } catch (IOException e) {
            ExceptionUtils.getMessage(e);
        }finally {
            lock.unlock();
        }
    }

    private static long pageOffset(int pageNumber) {
        return (long) (pageNumber - 1) * PAGE_SIZE;
    }
}
